package com.thp.bigdata.mapJoin;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.zookeeper.common.IOUtils;

/**
 * Map阶段实现join
 *   适用于关联表中有小表的情况
 *   Map端join是指数据到达map处理函数之前进行合并，效率要远高于Reduce端join
 *   因为Reduce端join是把所有的数据都经过shuffle，非常消耗资源
 *   
 *   
 * @author 汤小萌
 *
 */
public class TestDistributedCache {
	
	/**
	 * 什么是DistributeCache?
	 *   DistributeCache是为了方便用户进行应用程序开发而设计的文件分发工具。
	 *   它能够将只读的外部文件自动分发到各个节点上进行本地缓存，以便task运行时加载。
	 * 
	 */
	 
	
	/**
	 * 【注意:】在Map端join操作中，我们往往将小的表添加到内存中，因为内存的资源是非常宝贵的。
	 * 	这也说明了另一个问题，那就是如果表的数据量都非常大则不适合使用Map端join
	 */
	static class TestDistributedCacheMapper extends Mapper<LongWritable, Text, Text, Text> {
		FileReader in = null;
		BufferedReader reader = null;
		String localPath = null;
		
		HashMap<String, String> b_tab = new HashMap<String, String>();
		
		// map任务初始化的时候只调用一次
		@Override
		protected void setup(Context context)
				throws IOException, InterruptedException {
			// 获取到cache file 的本地绝对路径
			Path[] files = context.getLocalCacheFiles();
			localPath = files[0].toString();
			System.out.println("locaPath : " + localPath);
			URI[] cacheFiles = context.getCacheFiles();
			
			// 缓存文件的用法  -- 直接使用本地IO来读取
			in = new FileReader("b.txt");
			reader = new BufferedReader(in);
			String line = null;
			while(null != (line = reader.readLine())) {  // 读取一行
				String[] fields = line.split(",");
				b_tab.put(fields[0], fields[1]);
			}
			// 关流
			IOUtils.closeStream(reader);
			IOUtils.closeStream(in);
			
		}
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			// 这里读取的是这个map task 所负责的那一个切片数据(在hdfs上)
			String[] fields = value.toString().split("\t");
			String a_itemid = fields[0];
			String a_mount = fields[1];
			String b_name = b_tab.get(a_itemid);
			
			// 输出结果   
			context.write(new Text(a_itemid), new Text(a_mount + "\t" + " : " + localPath + "\t" + b_name));
			
		}
		
		
		
	}
	
	 
	
	public static void main(String[] args) throws IOException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.setJar("f:/rjoin.jar");
		
		job.setMapperClass(TestDistributedCacheMapper.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		
		
		
		
	}
	
	
}
